package com.JadePenG.demo;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;

import java.util.Arrays;
import java.util.Map;
import java.util.Random;

/**
 * BaseRichSpout 抽象方法
 * 打开数据源 读取一个文件
 */
public class ReadFileSpout extends BaseRichSpout {
    //private BufferedReader bufferedReader;
    private SpoutOutputCollector spoutOutputCollector;
    //增: 在这里定义了一个随机对象，用于获取随机数
    Random random;

    /**
     * 连接或者打开数据源, 类似于构造方法 旨在当前类被创建的时候 初始化一次
     * 打开之后永不关闭
     *
     * @param map                  strom集群
     * @param topologyContext      topology的上下文 一般不配置
     * @param spoutOutputCollector 收集器  收集数据,将数据封装到tuole, 将tuple发送给storm框架由storm框架向下游发送
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
       /* try {
            //storm通常不是和文件打交道 一般是和kafka
            this.bufferedReader = new BufferedReader(new FileReader(new File("h:/ReaderFile_title.txt")));
            this.spoutOutputCollector = spoutOutputCollector;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }*/

        //增: 初始化随机对象
        this.spoutOutputCollector = spoutOutputCollector;
        this.random = new Random();


    }

    /**
     * 底层是一个死循环 只要数据源有数据就一直读取, 主动将数据向下游发送
     * nextTuple 主动方法
     */
    @Override
    public void nextTuple() {
        /*try {
            //一行一行读取数据
            String line = bufferedReader.readLine();
            //不处理数据, 将读取的一行数据发走 将line封装到一个list中, 会封装到tuple里面,发送给下游Bolt
            //["i love hadoop"]  ["zhangsan","18',"男"]
            //有数据才读取
            if (line != null && !"".equals(line)) {
                spoutOutputCollector.emit(Arrays.asList(line));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }*/
        //增
        //随机从数组中拿去一句话
        String[] sentences = new String[]{"i love storm", "i love hadoop too", "i love spark too"};
        int index = random.nextInt(sentences.length);
        //一行一行的读取数据
        String line = sentences[index];
        try {
            Thread.sleep(1000);
            //将读取的一行数据发走  将line封装到一个list中，会封装到tuple里面，发送给下游Bolt
            if (line != null && !"".equals(line)) {
                spoutOutputCollector.emit(Arrays.asList(line), line);//["zhangsan",'18','男']
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 声明输出字段
     *
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //向下游发送的数据,进行定义  定义的字段要与发送的数据相对应
        //("name", "age", "sex")
        outputFieldsDeclarer.declare(new Fields("juzi"));
    }

    //是一个让下游调用的方法 如果接收到数据了下游就可以调用
    @Override
    public void ack(Object msgId) {
        System.out.println("接收到数据 哈哈哈" + msgId);
    }

    //是一个让下游调用的方法 如果下游出问题了，认为消息没接收到，就可以调用该方法
    @Override
    public void fail(Object msgId) {
        System.out.println("没消息呀  呜呜呜" + msgId);
        //collector.emit(msgId);可以重新发送失败的消息
    }
}
